package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.bean.PaymentWide;
import com.atguigu.realtime.bean.ProductStats;
import com.atguigu.realtime.function.DimAsyncFunction;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import com.atguigu.realtime.util.KKutil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.atguigu.realtime.common.Constant.*;

// RejectedExecutionException 因为端口重用,冲突了!
// Mailbox is in state CLOSED, but is required to be in state OPEN for put operations

//防止uv和uj长时间不更新 将水印推迟 但可能导致 uv和uj 数据迟到  解决方案：将乱序程度加大，一般超过5s（窗口长度），可设置为20s
// 水印加在开窗之前 15s的时候也是uj迟到。。。20s好像没有了
public class DwsProductStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsProductStatsApp().init(4002,1,"DwsProductStatsApp","DwsProductStatsApp",
                                                           TOPIC_DWD_PAGE, TOPIC_DWD_DISPLAY,
                                                           TOPIC_DWD_FAVOR_INFO, TOPIC_DWD_CART_INFO,
                                                           TOPIC_DWM_ORDER_WIDE, TOPIC_DWM_PAYMENT_WIDE,
                                                           TOPIC_DWD_ORDER_REFUND_INFO, TOPIC_DWD_COMMENT_INFO);
    }

    @Override
    protected void run(StreamExecutionEnvironment env,
                       HashMap<String, DataStreamSource<String>> topicToStream) {
        //1.把多个流union成一个流
        DataStream<ProductStats> psStream = parseAndUnionOne(topicToStream);
       // psStream.print();
        //2.开窗聚合
        SingleOutputStreamOperator<ProductStats> aggregatedStream = aggregate(psStream);
            // display  duo

        //3. join维度信息
        SingleOutputStreamOperator<ProductStats> psStreamWithDim = joinDim(aggregatedStream);
      //  psStreamWithDim.print();

        //4.数据写入到Clickhouse
        writeToClickHouse(psStreamWithDim);

        //5.写入到kafka中 (商品行为关键词主题需要)
        writeToKafka(psStreamWithDim);



    }

    private void writeToKafka(SingleOutputStreamOperator<ProductStats> psStreamWithDim) {
        psStreamWithDim
                .map(JSON::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(TOPIC_DWS_PRODUCT_STATS));
    }

    private void writeToClickHouse(SingleOutputStreamOperator<ProductStats> psStreamWithDim) {
        psStreamWithDim
                .addSink(FlinkSinkUtil
                .getClickHouseSink("gmall2021", "product_stats_2021", ProductStats.class));
    }

    private SingleOutputStreamOperator<ProductStats> joinDim(SingleOutputStreamOperator<ProductStats> stream) {
      return  AsyncDataStream.unorderedWait(
                stream,
                new DimAsyncFunction<ProductStats>() {
                    @Override
                    public void addDim(Jedis jedis,
                                       Connection phoenixConn,
                                       ProductStats ps,
                                       ResultFuture<ProductStats> resultFuture) {

                        // 3. 补充sku
                        JSONObject skuInfo = DimUtil.readDim(phoenixConn, jedis, "dim_sku_info", ps.getSku_id());
                        ps.setSku_name(skuInfo.getString("SKU_NAME"));
                        ps.setSku_price(skuInfo.getBigDecimal("PRICE"));

                        ps.setSpu_id(skuInfo.getLong("SPU_ID"));
                        ps.setTm_id(skuInfo.getLong("TM_ID"));
                        ps.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

                        // 4. 补充spu
                        JSONObject spuInfo = DimUtil.readDim(phoenixConn, jedis, "dim_spu_info", ps.getSpu_id());
                        ps.setSpu_name(spuInfo.getString("SPU_NAME"));
                        // 5. 补充tm
                        JSONObject tmInfo = DimUtil.readDim(phoenixConn, jedis, "dim_base_trademark", ps.getTm_id());
                        ps.setTm_name(tmInfo.getString("TM_NAME"));
                        // 5. 补充c3
                        JSONObject c3Info = DimUtil.readDim(phoenixConn, jedis, "dim_base_category3", ps.getCategory3_id());
                        ps.setCategory3_name(c3Info.getString("NAME"));
                        resultFuture.complete(Collections.singletonList(ps));
                    }
                },30,
                TimeUnit.SECONDS
        );
    }

    private SingleOutputStreamOperator<ProductStats> aggregate(DataStream<ProductStats> psStream) {
      return  psStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((vs,ts)->vs.getTs())
                )
                .keyBy(ProductStats::getSku_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats vs1, ProductStats vs2) throws Exception {

                                vs1.setClick_ct(vs1.getClick_ct()+vs2.getClick_ct());
                                vs1.setDisplay_ct(vs1.getDisplay_ct()+vs2.getDisplay_ct());
                                vs1.setCart_ct(vs1.getCart_ct()+vs2.getCart_ct());
                                vs1.setFavor_ct(vs1.getFavor_ct()+vs2.getFavor_ct());
                                vs1.setComment_ct(vs1.getComment_ct()+vs2.getComment_ct());
                                vs1.setGood_comment_ct(vs1.getGood_comment_ct()+vs2.getGood_comment_ct());

                                // BigDecimal格式用add 才行
                                vs1.setPayment_amount(vs1.getPayment_amount().add(vs2.getPayment_amount()));
                                vs1.setOrder_amount(vs1.getOrder_amount().add(vs2.getOrder_amount()));
                                vs1.setRefund_amount(vs1.getRefund_amount().add(vs2.getRefund_amount()));

                                vs1.setOrder_sku_num(vs1.getOrder_sku_num()+vs2.getOrder_sku_num());
                                vs1.getOrderIdSet().addAll(vs2.getOrderIdSet());
                                vs1.getRefundOrderIdSet().addAll(vs2.getRefundOrderIdSet());
                                vs1.getPaidOrderIdSet().addAll(vs2.getPaidOrderIdSet());
                                return vs1;
                            }
                        },
                        new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            @Override
                            public void process(Long key, Context ctx,
                                                Iterable<ProductStats> elements,
                                                Collector<ProductStats> out) throws Exception {
                                ProductStats ps = elements.iterator().next();
                                ps.setStt(KKutil.toDateTime(ctx.window().getStart()));
                                ps.setEdt(KKutil.toDateTime(ctx.window().getEnd()));
                                ps.setTs(System.currentTimeMillis()); //表示每天数据的统计时间

                                ps.setOrder_ct((long)ps.getOrderIdSet().size());
                                ps.setPaid_order_ct((long)ps.getPaidOrderIdSet().size());
                                ps.setRefund_order_ct((long)ps.getRefundOrderIdSet().size());
                                out.collect(ps);
                            }
                        });
    }

    private DataStream<ProductStats> parseAndUnionOne(HashMap<String, DataStreamSource<String>> topicToStream) {
        //1。点击
        SingleOutputStreamOperator<ProductStats> clickStream = topicToStream
                .get(TOPIC_DWD_PAGE)
                .flatMap(new FlatMapFunction<String, ProductStats>() {
                    @Override
                    public void flatMap(String value, Collector<ProductStats> out) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        JSONObject page = obj.getJSONObject("page");
                        String pageId = page.getString("page_id");
                        String itemType = page.getString("item_type");
                        if ("sku_id".equals(itemType) && pageId != null) {
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(page.getLong("item"));
                            ps.setClick_ct(1L);
                            //事件时间
                            ps.setTs(obj.getLong("ts"));
                            out.collect(ps);
                        }

                    }
                });
        //2.曝光
        //每一条都是曝光  但是 数据中   //曝光的是商品才行，因为以商品为粒度，有的是汉字
        SingleOutputStreamOperator<ProductStats> displayStream = topicToStream
                .get(TOPIC_DWD_DISPLAY)
                .filter(line -> "sku_id".equals(JSON.parseObject(line).getString("item_type")))
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    ProductStats ps = new ProductStats();

                    ps.setSku_id(obj.getLong("item"));
                    ps.setDisplay_ct(1L);
                    ps.setTs(obj.getLong("ts"));
                    return ps;
                });
        //3.收藏
        SingleOutputStreamOperator<ProductStats> favorStream = topicToStream
                .get(TOPIC_DWD_FAVOR_INFO)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setFavor_ct(1L);
                    ps.setTs(KKutil.toTs(obj.getString("create_time")));
                    return ps;
                });
        //4.购物车
        SingleOutputStreamOperator<ProductStats> cartStream = topicToStream
                .get(TOPIC_DWD_CART_INFO)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setCart_ct(1L);
                    ps.setTs(KKutil.toTs(obj.getString("create_time")));
                    return ps;
                });

        //5.订单宽表
        SingleOutputStreamOperator<ProductStats> orderWideStream = topicToStream
                .get(TOPIC_DWM_ORDER_WIDE)
                .map(line -> {
                    OrderWide orderWide = JSON.parseObject(line, OrderWide.class);

                    ProductStats ps = new ProductStats();
                    ps.setSku_id(orderWide.getSku_id());
                    ps.setOrder_amount(orderWide.getSplit_total_amount());
                    ps.setOrder_sku_num(orderWide.getSku_num());
                    ps.getOrderIdSet().add(orderWide.getOrder_id());
                    ps.setTs(KKutil.toTs(orderWide.getCreate_time()));

                    return ps;
                });
        //6.支付宽表
        SingleOutputStreamOperator<ProductStats> paymentStream = topicToStream
                .get(TOPIC_DWM_PAYMENT_WIDE)
                .map(line -> {
                    PaymentWide paymentWide = JSON.parseObject(line, PaymentWide.class);

                    ProductStats ps = new ProductStats();
                    ps.setSku_id(paymentWide.getSku_id());
                    ps.setPayment_amount(paymentWide.getTotal_amount());
                    ps.getPaidOrderIdSet().add(paymentWide.getOrder_id());
                    ps.setTs(KKutil.toTs(paymentWide.getPayment_create_time()));

                    return ps;
                });
        //7.退款
        SingleOutputStreamOperator<ProductStats> refundStream = topicToStream
                .get(TOPIC_DWD_ORDER_REFUND_INFO)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setRefund_amount(obj.getBigDecimal("refund_amount"));
                    ps.getRefundOrderIdSet().add(obj.getLong("order_id"));
                    ps.setTs(KKutil.toTs(obj.getString("create_time")));
                    return ps;
                });
        //8.评价
        SingleOutputStreamOperator<ProductStats> commentStream = topicToStream
                .get(TOPIC_DWD_COMMENT_INFO)
                .map(line -> {
                    JSONObject obj = JSON.parseObject(line);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setComment_ct(1L);
                    String appraise = obj.getString("appraise");
                    if (FOUR_STAR_COMMENT.equals(appraise) ||FIVE_STAR_COMMENT.equals(appraise)){
                        ps.setGood_comment_ct(1L);
                    }
                    ps.setTs(KKutil.toTs(obj.getString("create_time")));
                    return ps;
                });
        return  clickStream.union(favorStream,
                                  displayStream,
                                  cartStream,
                                  refundStream,
                                  orderWideStream,
                                  paymentStream,
                                  commentStream);
    }

}
